/*
* This file is a part of the open-eBackup project.
* This Source Code Form is subject to the terms of the Mozilla Public License, v. 2.0.
* If a copy of the MPL was not distributed with this file, You can obtain one at
* http://mozilla.org/MPL/2.0/.
*
* Copyright (c) [2024] Huawei Technologies Co.,Ltd.
*
* THIS SOFTWARE IS PROVIDED ON AN "AS IS" BASIS, WITHOUT WARRANTIES OF ANY KIND,
* EITHER EXPRESS OR IMPLIED, INCLUDING BUT NOT LIMITED TO NON-INFRINGEMENT,
* MERCHANTABILITY OR FIT FOR A PARTICULAR PURPOSE.
*/
#include <sstream>
#include <protect_engines/kubernetes/rest/config/KubeConfig.h>
#include <protect_engines/kubernetes/rest/client/KubeClient.h>
#include <protect_engines/kubernetes/rest/KubernetesApi.h>
#include <protect_engines/kubernetes/common/KubeErrorCodes.h>
#include <protect_engines/kubernetes/common/KubeTaskLabels.h>
#include <protect_engines/kubernetes/util/Transformer.h>
#include <volume_handlers/oceanstor/OceanStorVolumeHandler.h>
#include <boost/lexical_cast.hpp>
#include "volume_handlers/oceanstor/DiskScannerHandler.h"
#include "yaml-cpp/yaml.h"
#include "common/JsonHelper.h"
#include "log/Log.h"
#include "common/Structs.h"
#include "job_controller/jobs/backup/BackupJob.h"
#include "common/Structs.h"
#include "KubernetesProtectEngine.h"
#include "common/utils/Utils.h"

namespace {
const std::string MODULE_NAME = "KubernetesProtectEngine";
const std::string EXTEND_INFO_INVALID = "AppEnv auth extendInfo parse error.";
const std::string STORAGES_OR_CONFIG_EMPTY = "The storages or config are empty.";
const std::string CONFIG_CONTENT_INVALID = "The config content is invalid.";
const std::string K8S_API_CONNECT_FAIL = "Fail to connect k8s cluster.";
const std::string DORADO_AUTH_FAIL = "Fail to connect storage or fail to auth storage.";
const std::string SNAPSHOT_ACTIVE_STATUS = "43";

const int K8S_RETRY_TIMES = 30; // 30 次
const int K8S_WAIT_PERIOD = 20; // 间隔时间 20 s
const int THREE_MINUTES = 180; // 180s
const int NUM_100 = 100;
const bool PRINT_LABEL = true;
const bool NOT_PRINT_LABEL = false;
const int32_t DEFAULT_STORAGE_LIMIT = 10;
const std::string REPORT_COPY_SUB_JOB = "ReportCopySubJob";
const std::string SUDO_DISK_TOOL_PATH = VirtPlugin::VIRT_PLUGIN_PATH + "bin/security_sudo_disk.sh";
}

namespace KubernetesPlugin {
int32_t KubernetesProtectEngine::CreateSnapshot(SnapshotInfo &snapshot, std::string & /* errCode */)
{
    INFOLOG("Start to create PVs snapshot.");
    std::vector <BatchSnapshotParam> createParams = {};
    int ret = ParseBackupPvsAndStorage(createParams, PRINT_LABEL);
    if (ret != SUCCESS) {
        ERRLOG("Parse pv list and storage fail.");
        return FAILED;
    }
    // key：存储的ESN，Value:一致性快照信息
    std::map <std::string, ConsistentActivationInfo> esnMap;
    for (auto &createParam: createParams) {
        ret = TakeConsistentSnapshots(createParam, esnMap);
        if (ret != SUCCESS) {
            ERRLOG("Batch consistent snapshot failed.");
            return FAILED;
        }
        GenerateVolSnapInfo(createParam, snapshot);
    }
    // 对同一存储设备上的快照进行一致性激活
    ret = ActiveConsistentSnapshots(esnMap);
    if (ret != SUCCESS) {
        ERRLOG("Active consistent snapshot failed.");
        return FAILED;
    }
    m_snapshotCache = createParams;
    m_metaDataCached = true;
    return SUCCESS;
}

int32_t KubernetesProtectEngine::PreDeleteSnapshot(const VirtPlugin::VolSnapInfo &item,
    std::shared_ptr<StorageClient> storageClient)
{
    auto [querySnapshotRet, snapshotInfoData] = storageClient->QuerySnapshot(item.m_snapshotId);
    if (querySnapshotRet != SUCCESS) {
        ERRLOG("Query snapshot status failed when delete snapshot,snapshot ID: %s,snapshot name: %s,error code: %d",
            item.m_snapshotId.c_str(), item.m_snapshotName.c_str(), querySnapshotRet);
        return FAILED;
    }
    // 判断快照状态是否是已激活，若已激活，则取消激活
    if (snapshotInfoData.m_runningStatus == SNAPSHOT_ACTIVE_STATUS) {
        int stopRet = storageClient->StopSnapshot(snapshotInfoData.m_id);
        if (stopRet != SUCCESS) {
            ERRLOG("Failed to stop snapshot, snapshot name: %s, snapshot ID: %s, error code: %d",
                snapshotInfoData.m_name.c_str(), snapshotInfoData.m_id.c_str(), stopRet);
            return FAILED;
        }
    }
    return SUCCESS;
}


int32_t KubernetesProtectEngine::DeleteSnapshot(const SnapshotInfo &snapshot)
{
    std::shared_ptr<AppProtect::BackupJob> backupParam = GetBackupParam();
    auto storageClients = KubeHelper::FindMatchSnapInfoStorageClients(snapshot, backupParam);
    if (storageClients.empty()) {
        ERRLOG("Failed to generate storage client when delete snaps, task ID: %s", GetBackupParam()->jobId);
        return FAILED;
    }
    bool deleteFailed = false;
    for (const auto &item : snapshot.m_volSnapList) {
        std::shared_ptr<StorageClient> storageClient = storageClients[item.m_datastore.m_moRef];
        if (storageClient == nullptr) {
            ERRLOG("Failed to obtain the client corresponding to the current snapshot,delete snapshot failed.\
                snapshot ID: %s,snapshot name: %s", item.m_snapshotId.c_str(), item.m_snapshotName.c_str());
            continue;
        }
        if (PreDeleteSnapshot(item, storageClient) != SUCCESS) {
            ERRLOG("Failed to PreDeleteSnapshot. snapshot ID: %s,snapshot name: %s", item.m_snapshotId.c_str(),
                item.m_snapshotName.c_str());
            continue;
        }
        // 检查快照关联Lun组情况，若关联Lun组，尝试移除
        int perRet = storageClient->RemoveSnapshotFromLunGroupExp(item.m_snapshotId);
        if (perRet != SUCCESS) {
            ERRLOG("Remove snapshot from LunGroup failed, snapshot ID: %s, snapshot name: %s, response code: %d",
                item.m_snapshotId.c_str(), item.m_snapshotName.c_str(), perRet);
        }
        perRet = storageClient->DeleteSnapshot(item.m_snapshotId);
        if (perRet == SNAP_NOT_EXIST) {
            INFOLOG("The snapshot does not exist, snapshot ID: %s, snapshot name: %s, response code: %d",
                item.m_snapshotId.c_str(), item.m_snapshotName.c_str(), perRet);
            continue;
        }
        if (perRet != SUCCESS) {
            ERRLOG("The snapshot deleted failed, snapshot ID: %s, snapshot name: %s, response code: %d",
                item.m_snapshotId.c_str(), item.m_snapshotName.c_str(), perRet);
            GenerateLabel(JobLogLevel::type::TASK_LOG_WARNING, DELETE_SNAP_FAIL_LABEL,
                          std::vector{item.m_snapshotName});
            deleteFailed = true;
            continue;
        }
        GenerateLabel(JobLogLevel::type::TASK_LOG_INFO, DELETE_SNAP_SUCCESS_LABEL, std::vector{item.m_snapshotName});
    }
    if (deleteFailed) {
        return FAILED;
    }
    return SUCCESS;
}

int32_t KubernetesProtectEngine::PostScriptTaskForRestore(const int32_t jobExecResult)
{
    std::shared_ptr<AppProtect::RestoreJob> restoreJobPtr = GetRestoreParam();
    if (jobExecResult == SUCCESS) {
        auto postScriptOpt = KubeHelper::GetScriptFromExtendInfo(restoreJobPtr->targetObject.extendInfo,
                                                                 ScriptType::POST_SCRIPT);
        if (!postScriptOpt || postScriptOpt.value().empty()) {
            return SUCCESS;
        }
        INFOLOG("Ready to run postScript: %s", postScriptOpt.value().c_str());
        sleep(THREE_MINUTES);
        RunScriptOnAllPodsInTheStateFulSet(postScriptOpt.value(), restoreJobPtr, TaskStage::Post_Task);
    } else {
        auto failedScriptOpt = KubeHelper::GetScriptFromExtendInfo(restoreJobPtr->targetObject.extendInfo,
                                                                   ScriptType::FAILED_SCRIPT);
        if (!failedScriptOpt || failedScriptOpt.value().empty()) {
            return SUCCESS;
        }
        INFOLOG("Ready to run failedScript: %s", failedScriptOpt.value().c_str());
        sleep(THREE_MINUTES);
        RunScriptOnAllPodsInTheStateFulSet(failedScriptOpt.value(), restoreJobPtr, TaskStage::Fail_Task);
    }
    return SUCCESS;
}

int32_t KubernetesProtectEngine::PostScriptTaskForBackup(const int32_t jobExecResult)
{
    std::shared_ptr<AppProtect::BackupJob> backupParam = GetBackupParam();
    BackupJobParamExtendInfo backupJobParamExtendInfo;
    if (!Module::JsonHelper::JsonStringToStruct(backupParam->protectObject.extendInfo, backupJobParamExtendInfo)) {
        ERRLOG("Protect object extend info parse fail, protectObject.extendInfo: %s",
            backupParam->protectObject.extendInfo.c_str());
        return FAILED;
    }

    if (jobExecResult == SUCCESS && !backupJobParamExtendInfo.m_postScript.empty()) {
        INFOLOG("Job success, execute post script.");
        BackupPostHook(backupJobParamExtendInfo.m_postScript, TaskStage::Post_Task);
    }

    if (jobExecResult != SUCCESS && !backupJobParamExtendInfo.m_failedScript.empty()) {
        INFOLOG("Job failed, execute failure script.");
        BackupPostHook(backupJobParamExtendInfo.m_failedScript, TaskStage::Fail_Task);
    }

    return SUCCESS;
}

int32_t KubernetesProtectEngine::PostHook(const VirtPlugin::ExecHookParam &para)
{
    int32_t ret = SUCCESS;
    DBGLOG("KubernetesProtectEngine post hook, %d", para.jobExecRet);
    JobType jobType = GetJobHandle()->GetJobType();
    if (jobType == JobType::BACKUP && para.stage == VirtPlugin::JobStage::POST_JOB &&
        para.hookType == VirtPlugin::HookType::POST_HOOK) {
        return PostScriptTaskForBackup(para.jobExecRet);
    }  else if (jobType == JobType::RESTORE && para.stage == VirtPlugin::JobStage::POST_JOB &&
                para.hookType == VirtPlugin::HookType::POST_HOOK) {
        return PostScriptTaskForRestore(para.jobExecRet);
    }

    INFOLOG("Finish Post Hook!");
    return SUCCESS;
}

int32_t KubernetesProtectEngine::PreHook(const VirtPlugin::ExecHookParam &para)
{
    if (para.stage == VirtPlugin::JobStage::POST_JOB) {
        DBGLOG("Post stage, no need prehook");
        return SUCCESS;
    }
    std::vector<std::string> cmdOut;
    std::string agentHomedir = Module::EnvVarManager::GetInstance()->GetAgentHomePath();
    std::vector<Module::CmdParam> cmdParam {
        Module::CmdParam(Module::COMMON_CMD_NAME, "sudo"),
        Module::CmdParam(Module::SCRIPT_CMD_NAME, agentHomedir + SUDO_DISK_TOOL_PATH),
        "get_scsi_list"
    };
    if (VirtPlugin::Utils::CallAgentExecCmd(cmdParam, cmdOut) != 0) {
        ERRLOG("lsscsi command is not enable!");
        GenerateLabel(JobLogLevel::type::TASK_LOG_ERROR, LACK_NEED_TOOL_LABEL,
            std::vector<std::string>{"lsscsi"});
        return Module::FAILED;
    }
    return SUCCESS;
}

std::pair<int32_t, std::map<std::string, std::string>> KubernetesProtectEngine::RunScriptOnAllPodsInTheStateFulSet(
    const std::string &scriptName, const std::shared_ptr<AppProtect::RestoreJob> &restoreJobPtr, TaskStage stage)
{
    int32_t ret = FAILED;
    std::map<std::string, std::string> podsExecuteResult =  {};

    auto targetStateFulSet = KubeHelper::GetStateFulSetFromExtendInfo(restoreJobPtr->targetObject.extendInfo);
    if (!targetStateFulSet) {
        return make_pair(ret, podsExecuteResult);
    }

    auto kubernetesApiOpt = KubeHelper::GetKubernetesApiFromAppEnv(restoreJobPtr->targetEnv.auth.extendInfo);
    if (!kubernetesApiOpt) {
        return make_pair(ret, podsExecuteResult);
    }

    auto [result, targetPods] = kubernetesApiOpt->ListPods(restoreJobPtr->targetObject);
    if (result != SUCCESS || targetPods.empty()) {
        ERRLOG("Target StateFulSet is unHealthy when doing RunScript");
        return make_pair(ret, podsExecuteResult);
    }
    for (const auto& pod : targetPods) {
        PodExtendInfo podExtendInfo;
        if (!Module::JsonHelper::JsonStringToStruct(pod.extendInfo, podExtendInfo)) {
            ERRLOG("Struct PodExtendInfo to string fail.");
            continue;
        }
        for (const auto& containerName : podExtendInfo.m_containerNames) {
            auto cmdResult = kubernetesApiOpt->KubeExec(targetStateFulSet->nameSpace, pod.name,
                                                        containerName, scriptName);
            INFOLOG("Execute script result in container pod, pod name: %s, container name: %s, exec ret:%s",
                pod.name.c_str(), containerName.c_str(), cmdResult.c_str());
            GenerateScriptLabel(cmdResult, pod.name, containerName, scriptName, stage);
            podsExecuteResult.insert(make_pair(pod.name, cmdResult));
        }
    }
    return make_pair(ret, podsExecuteResult);
}

int32_t KubernetesProtectEngine::BackupPostHook(const std::string &scriptPath, TaskStage stage)
{
    std::shared_ptr<AppProtect::BackupJob> backupParam = GetBackupParam();

    auto kubernetesApi = KubeHelper::GetKubernetesApiFromAppEnv(backupParam->protectEnv.auth.extendInfo);
    if (!kubernetesApi.has_value()) {
        ERRLOG("Failed to initialize kubernetesApi, protectObject.extendInfo: %s",
            backupParam->protectObject.extendInfo.c_str());
        return FAILED;
    }

    Application statefulSet = backupParam->protectObject;
    auto[ret, pods] = kubernetesApi->ListPods(statefulSet);
    if (ret != SUCCESS) {
        ERRLOG("Fail to extract authExtendInfo.");
        return FAILED;
    }

    if (pods.empty()) {
        ERRLOG("Post hook, do not find any pod in charge under the stateful set, stateful set name: %s",
            statefulSet.name.c_str());
        return SUCCESS;
    }

    for (const auto &pod : pods) {
        PodExtendInfo podExtendInfo;
        if (!Module::JsonHelper::JsonStringToStruct(pod.extendInfo, podExtendInfo)) {
            ERRLOG("Struct PodExtendInfo to string fail.");
            continue;
        }

        for (const auto &containerName : podExtendInfo.m_containerNames) {
            std::string excResult = kubernetesApi->KubeExec(pod.parentName, pod.name, containerName,
                                                            scriptPath);
            INFOLOG("Post backup hook, Execute script result in container pod, result: %s", excResult.c_str());
            GenerateScriptLabel(excResult, pod.name, containerName, scriptPath, stage);
        }
    }
    return SUCCESS;
}

int32_t KubernetesProtectEngine::QuerySnapshotExists(SnapshotInfo &snapshotInfo)
{
    std::shared_ptr<AppProtect::BackupJob> backupParam = GetBackupParam();
    auto storageClients = KubeHelper::FindMatchSnapInfoStorageClients(snapshotInfo, backupParam);
    if (storageClients.empty()) {
        ERRLOG("Failed to generate storage client when query snaps, task ID: %s", GetBackupParam()->jobId.c_str());
        return FAILED;
    }

    snapshotInfo.m_deleted = true;
    for (auto &item : snapshotInfo.m_volSnapList) {
        item.m_deleted = true;
        std::shared_ptr<StorageClient> storageClient = storageClients[item.m_datastore.m_moRef];
        if (storageClient == nullptr) {
            ERRLOG("Failed to obtain the client corresponding to the current snapshot,snapshot ID:%s ,\
                snapshot name: %s", item.m_snapshotId.c_str(), item.m_snapshotName.c_str());
            continue;
        }
        auto[perRet, snapshotData] = storageClient->QuerySnapshot(item.m_snapshotId);
        if (perRet == SUCCESS) {
            ERRLOG("The Snapshot still exists, snapshot ID: %s snapshot name: %s",
                item.m_snapshotId.c_str(), item.m_snapshotName.c_str());
            snapshotInfo.m_deleted = false;
            item.m_deleted = false;
        }
    }

    return SUCCESS;
}

int32_t KubernetesProtectEngine::GetPvcLunInfo(BatchSnapshotParam &createParam)
{
    auto [ret, storageClient] = KubeHelper::GetStorageClientFromStorageParam(createParam.m_storage);
    if (ret != SUCCESS) {
        ERRLOG("Storage device authentication failed, ip:%s, username:%s, error code: %d",
            createParam.m_storage.ip.c_str(), createParam.m_storage.username.c_str(), ret);
        return FAILED;
    }
    StorageDeviceInfo storageDeviceInfo;
    std::tie(ret, storageDeviceInfo) = storageClient->GetDeviceBaseInfo();
    if (ret != SUCCESS) {
        ERRLOG("Failed to obtain basic information of DeviceBaseInfo, ip:%s, username:%s, error code: %d",
            createParam.m_storage.ip.c_str(), createParam.m_storage.username.c_str(), ret);
        return FAILED;
    }
    createParam.m_storageDeviceInfo = storageDeviceInfo;
    std::vector <PerSnapshotParam> snapshotParams = {};
    for (const auto &pv: createParam.m_pvs) {
        PerSnapshotParam perSnapshotParam;
        perSnapshotParam.m_pv = pv;
        // 通过卷名称获取卷ID
        auto[perRet, lunInfoData] = storageClient->GetLunInfoData(pv.lunName);
        if (perRet != SUCCESS) {
            ERRLOG("Failed to get volume information from storage device, LUN name: %s, error code: %d",
                pv.lunName.c_str(), perRet);
            return FAILED;
        }
        perSnapshotParam.m_lunInfoData = lunInfoData;
        snapshotParams.push_back(perSnapshotParam);
    }
    createParam.m_snapshotInfos = snapshotParams;
    return SUCCESS;
}

int32_t KubernetesProtectEngine::GetMachineMetadata(VMInfo &vmInfo)
{
    if (!m_metaDataCached) {
        std::vector <BatchSnapshotParam> createParams = {};
        int32_t iRet = ParseBackupPvsAndStorage(createParams, NOT_PRINT_LABEL);
        if (iRet != SUCCESS) {
            ERRLOG("Parse pv list and storage fail.");
            return FAILED;
        }
        for (auto &createParam: createParams) {
            if (GetPvcLunInfo(createParam) != SUCCESS) {
                ERRLOG("Parse pv from pod(%s) luninfo fail. %s",
                    createParam.m_pod.name.c_str(), m_jobId.c_str());
                continue;
            }
        }
        m_snapshotCache = createParams;
    }
    std::shared_ptr<AppProtect::BackupJob> backupParam = GetBackupParam();
    std::vector<VolInfo> volInfos;
    for (auto &batchSnapshotParam : m_snapshotCache) {
        for (const auto &snapshotInfo : batchSnapshotParam.m_snapshotInfos) {
            auto[ret, volInfo]  = KubeHelper::GenerateVolInfo(snapshotInfo, backupParam, batchSnapshotParam);
            if (ret != SUCCESS) {
                ERRLOG("SnapshotInfo transfer to json string err.");
                return FAILED;
            }
            volInfos.push_back(volInfo);
        }
    }
    vmInfo.m_volList = volInfos;
    vmInfo.m_moRef = backupParam->protectObject.id;
    vmInfo.m_uuid = backupParam->protectObject.id;
    vmInfo.m_name = backupParam->protectObject.name;
    vmInfo.m_location = backupParam->protectEnv.id;
    return SUCCESS;
}

int32_t KubernetesProtectEngine::GetVolumesMetadata(const VMInfo& /* vmMetadata */,
    std::unordered_map<std::string, std::string>& /* volsMetadata */)
{
    return SUCCESS;
}

int32_t KubernetesProtectEngine::GetVolumeHandler(const VolInfo & volInfo,
    std::shared_ptr<VolumeHandler>& volHandler)
{
    std::shared_ptr<VirtPlugin::OceanStorVolumeHandler> storageVolHandler =
        std::make_shared<VirtPlugin::OceanStorVolumeHandler>(GetJobHandle(), volInfo, m_jobId, m_subJobId);
    if (!m_noNeedInitOceanVolHandler && storageVolHandler->InitializeVolumeInfo() != SUCCESS) {
        ERRLOG("Initialize volume info failed.");
        return FAILED;
    }
    volHandler = storageVolHandler;
    return SUCCESS;
}

int32_t KubernetesProtectEngine::CreateVolume(const VolInfo &volObj, const std::string &volMetaData,
                                              const std::string &vmMoRef, const DatastoreInfo &dsInfo, VolInfo &newVol)
{
    return SUCCESS;
}

int32_t KubernetesProtectEngine::DetachVolume(const VolInfo &volObj)
{
    return SUCCESS;
}

int32_t KubernetesProtectEngine::AttachVolume(const VolInfo &volObj)
{
    return SUCCESS;
}

int32_t KubernetesProtectEngine::DeleteVolume(const VolInfo &volObj)
{
    return SUCCESS;
}

int32_t KubernetesProtectEngine::ReplaceVolume(const VolInfo &volObj)
{
    return SUCCESS;
}

int32_t KubernetesProtectEngine::CreateMachine(VMInfo& /* vmInfo */)
{
    return SUCCESS;
}

int32_t KubernetesProtectEngine::DeleteMachine(const VMInfo& /* vmInfo */)
{
    return SUCCESS;
}

int32_t KubernetesProtectEngine::RenameMachine(const VMInfo& /* vmInfo */, const std::string& /* newName */)
{
    return SUCCESS;
}

int32_t KubernetesProtectEngine::PowerOnMachine(const VMInfo& /* vmInfo */)
{
    std::shared_ptr<AppProtect::RestoreJob> restoreJobPtr = GetRestoreParam();
    auto targetStateFulSet = KubeHelper::GetStateFulSetFromExtendInfo(restoreJobPtr->targetObject.extendInfo);
    if (!targetStateFulSet) {
        return FAILED;
    }
    if (targetStateFulSet->replicasNum == 0) {
        WARNLOG("The target's replicasNum is already zero! No need to Power on.");
        return SUCCESS;
    }

    auto kubernetesApiOpt = KubeHelper::GetKubernetesApiFromAppEnv(restoreJobPtr->targetEnv.auth.extendInfo);
    if (!kubernetesApiOpt) {
        return FAILED;
    }
    auto ret = kubernetesApiOpt->RestoreStateFulSet(*targetStateFulSet);
    if (ret != SUCCESS) {
        return FAILED;
    }
    // 恢复stateFulSet后， 尝试查询pod，直到statefulset下面的pod数恢复
    int retryTimes = 0;
    std::vector<ApplicationResource> pods = {};
    do {
        std::tie(ret, pods) = kubernetesApiOpt->ListPods(restoreJobPtr->targetObject);
        if (ret == SUCCESS && pods.size() == targetStateFulSet->replicasNum || ret != SUCCESS) {
            INFOLOG("Finish Powering On with ret:%d pods size: %d", ret, pods.size());
            return ret;
        }
        retryTimes++;
        INFOLOG("try connect pod after set replicas back to: %d retry times: %d", targetStateFulSet->replicasNum,
            retryTimes);
        sleep(K8S_WAIT_PERIOD);
    } while (retryTimes < K8S_RETRY_TIMES);
    return SUCCESS;
}

int32_t KubernetesProtectEngine::PowerOffMachine(const VMInfo& vmInfo)
{
    std::shared_ptr<AppProtect::RestoreJob> restoreJobPtr = GetRestoreParam();
    auto targetStateFulSet = KubeHelper::GetStateFulSetFromExtendInfo(restoreJobPtr->targetObject.extendInfo);
    if (!targetStateFulSet) {
        return FAILED;
    }
    if (targetStateFulSet->replicasNum == 0) {
        WARNLOG("The target's replicasNum is already zero! No need to Power Off");
        return SUCCESS;
    }
    auto kubernetesApiOpt = KubeHelper::GetKubernetesApiFromAppEnv(restoreJobPtr->targetEnv.auth.extendInfo);
    if (!kubernetesApiOpt) {
        return FAILED;
    }
    auto ret = kubernetesApiOpt->StopStateFulSet(targetStateFulSet.value());
    if (ret != SUCCESS) {
        return FAILED;
    }
    // 关stateFulSet后， 尝试连接相关pod，直到statefulset下面的pod数为0
    int retryTimes = 0;
    std::vector<ApplicationResource> pods = {};
    do {
        std::tie(ret, pods) = kubernetesApiOpt->ListPods(restoreJobPtr->targetObject);
        if (ret == SUCCESS && pods.empty() || ret != SUCCESS) {
            INFOLOG("Finish Powering Off with ret:%d pods size: %d", ret, pods.size());
            return ret;
        }
        retryTimes++;
        INFOLOG("try connect pod after set replicas to 0, retry times: %d", retryTimes);
        sleep(K8S_WAIT_PERIOD);
    } while (retryTimes < K8S_RETRY_TIMES);
    ERRLOG("try to set replicas to zero failed,power off machine failed");
    GenerateLabel(JobLogLevel::type::TASK_LOG_ERROR, SCALE_THE_PODS_TO_ZERO_FAILED_LABEL,
                  std::vector<std::string>{});
    return FAILED;
}

int32_t KubernetesProtectEngine::AllowBackupInLocalNode(const AppProtect::BackupJob &job, int32_t &errorCode)
{
    // 检查插件与k8s集群的连通性是否成功
    auto ret = CheckProtectEnvConn(job.protectEnv);
    if (ret != SUCCESS) {
        return FAILED;
    }
    // 检查插件与生产存储的连通性
    return CheckAllStorageConnection(job.protectEnv.auth.extendInfo);
}

int32_t KubernetesProtectEngine::AllowRestoreInLocalNode(const AppProtect::RestoreJob &job, int32_t &errorCode)
{
    // 检查插件与k8s集群的连通性是否成功
    auto ret = CheckProtectEnvConn(job.targetEnv);
    if (ret != SUCCESS) {
        return FAILED;
    }
    // 检查插件与生产存储的连通性
    return CheckAllStorageConnection(job.targetEnv.auth.extendInfo);
}

int32_t KubernetesProtectEngine::AllowBackupSubJobInLocalNode(const AppProtect::BackupJob &job,
                                                              const AppProtect::SubJob &subJob, int32_t &errorCode)
{
    if (CheckProtectEnvConn(job.protectEnv) != SUCCESS) {
        return FAILED;
    }
    if (subJob.jobName == REPORT_COPY_SUB_JOB) {
        return SUCCESS;
    }
    if (subJob.jobType == SubJobType::type::POST_SUB_JOB) {
        return SUCCESS;
    }
    if (CheckAgentEnv() != SUCCESS) {
        return FAILED;
    }
    VirtPlugin::BackupSubJobInfo backupSubJob{};
    if (!Module::JsonHelper::JsonStringToStruct(subJob.jobInfo, backupSubJob)) {
        ERRLOG("Get backup subjob info failed");
        return FAILED;
    }
    std::string authExtendInfo = job.protectEnv.auth.extendInfo;
    return CheckStorageConnection(backupSubJob.m_volInfo, authExtendInfo);
}

int32_t KubernetesProtectEngine::AllowRestoreSubJobInLocalNode(const AppProtect::RestoreJob &job,
                                                               const AppProtect::SubJob &subJob, int32_t &errorCode)
{
    if (CheckProtectEnvConn(job.targetEnv) != SUCCESS) {
        return FAILED;
    }
    if (subJob.jobType == SubJobType::type::POST_SUB_JOB) {
        return SUCCESS;
    }
    if (subJob.jobName == "PostSubJob") {
        return SUCCESS;
    }
    if (CheckAgentEnv() != SUCCESS) {
        return FAILED;
    }
    VirtPlugin::SubJobExtendInfo subJobExtendInfo{};
    if (!Module::JsonHelper::JsonStringToStruct(subJob.jobInfo, subJobExtendInfo)) {
        ERRLOG("Get restore subjob info failed.");
        return FAILED;
    }
    VolInfo volInfo{};
    if (!Module::JsonHelper::JsonStringToStruct(subJobExtendInfo.m_targetVolumeInfo, volInfo)) {
        ERRLOG("Failed to get vol info");
        return FAILED;
    }
    std::string authExtendInfo = job.targetEnv.auth.extendInfo;
    return CheckStorageConnection(volInfo, authExtendInfo);
}

int32_t KubernetesProtectEngine::CheckProtectEnvConn(const AppProtect::ApplicationEnvironment &env)
{
    // 校验k8sApi输入参数，如果成功得到k8sApi, 并查看FlexVolume连通性
    auto kubernetesApi = KubeHelper::GetKubernetesApiFromAppEnv(env.auth.extendInfo);
    if (!kubernetesApi) {
        ERRLOG("Current agent can not connect kubernetes produce environment");
        return FAILED;
    }
    auto [ret, nameSpaceResourceList] = kubernetesApi->ListNameSpaces();
    if (ret != SUCCESS) {
        ERRLOG("Current agent can not connect kubernetes produce environment");
        return FAILED;
    }
    return SUCCESS;
}

int32_t KubernetesProtectEngine::CheckStorageConnection(const VolInfo &volInfo, const std::string &authExtendInfo)
{
    std::shared_ptr <VolumeHandler> storageVolHandler = nullptr;
    m_noNeedInitOceanVolHandler = true;
    if (GetVolumeHandler(volInfo, storageVolHandler) != SUCCESS) {
        ERRLOG("Get volume handler failed when check storage connection");
        return FAILED;
    }
    int32_t erroCode;
    if (storageVolHandler->TestDeviceConnection(authExtendInfo, erroCode) != SUCCESS) {
        ERRLOG("Get volume handler failed when check storage connection");
        return FAILED;
    }
    DBGLOG("Connect to storage success");
    return SUCCESS;
}

int32_t KubernetesProtectEngine::CheckAgentEnv()
{
    if (!VirtPlugin::DiskScannerHandler::GetInstance()->IsInstallIscsiInitiator()) {
        ERRLOG("The iSCSI initiator is not installed on the current host or the iSCSI service is not started.");
        return Module::FAILED;
    }
    DBGLOG("Current host installed iscsi initiator");
    return SUCCESS;
}

int32_t KubernetesProtectEngine::CheckAllStorageConnection(const std::string &authExtendInfo)
{
    // 获取生产存储参数列表
    auto storageParamList = KubeHelper::GetStorageParamVecFromAppEnv(authExtendInfo);
    if (!storageParamList || storageParamList->size() == 0) {
        ERRLOG("%s", STORAGES_OR_CONFIG_EMPTY);
        return FAILED;
    }
    // 检查所有存储的连通性
    for (const auto &item: storageParamList.value()) {
        std::string storageInputUrl = "https://" + item.ip + ":" + std::to_string(item.port) + "/deviceManager/rest";
        KubernetesPlugin::AccessAuthParam accessAuthParam(item.username, item.password, "0");
        auto [result, storageClient] = StorageClient::Create(item.ip, item.port, accessAuthParam, item.ipList);
        if (result != 0) {
            int errCode = (result == USERNAME_OR_PASSWORD_WRONG_V6 || result == USERNAME_OR_PASSWORD_WRONG_V3) ?
                          STORAGE_AUTH_FAILED : STORAGE_NETWORK_FAILED;
            ERRLOG("%s Storage url: %s Storage username: %s ErrorCode: %d", DORADO_AUTH_FAIL.c_str(),
                   storageInputUrl.c_str(), item.username.c_str(), errCode);
            return FAILED;
        }
    }
    return SUCCESS;
}

int32_t KubernetesProtectEngine::CheckBackupJobType(const VirtPlugin::JobTypeParam& jobTypeParam, bool& checkRet)
{
    auto storagesOpt = KubeHelper::GetStorageParamVecFromAppEnv(jobTypeParam.m_job.protectEnv.auth.extendInfo);
    if (!storagesOpt) {
        ERRLOG("Generate storages from backup param fail, task ID: %s", jobTypeParam.m_job.jobId.c_str());
        return FAILED;
    }

    std::string storageSn = jobTypeParam.m_snapshotInfo.m_volSnapList.back().m_datastore.m_moRef;
    if (storageSn.empty()) {
        ERRLOG("Failed to get sn from pre snapshot list, task ID: %s", jobTypeParam.m_job.jobId.c_str());
        return FAILED;
    }

    auto storageOpt = KubeHelper::FindMatchSnStorage(storagesOpt.value(), storageSn);
    if (!storageOpt.has_value()) {
        ERRLOG("No usable storage device param found in backup param, required sn: %s, task ID: %s",
            storageSn.c_str(), jobTypeParam.m_job.jobId.c_str());
        return FAILED;
    }
    StorageParam storage = storageOpt.value();

    auto[initRet, storageClient] = KubeHelper::GetStorageClientFromStorageParam(storage);
    if (initRet != SUCCESS) {
        ERRLOG("Storage auth fail, task ID: %s, ip: %s, port: %d, response code: %d",
            jobTypeParam.m_job.jobId.c_str(), storage.ip.c_str(), storage.port, initRet);
        return FAILED;
    }

    checkRet = true;
    for (auto &item : jobTypeParam.m_snapshotInfo.m_volSnapList) {
        auto[perRet, snapshotData] = storageClient->QuerySnapshot(item.m_snapshotId);
        if (perRet != SUCCESS) {
            ERRLOG("The pre snapshot not exists, task ID: %s, snapshot ID: %s, snapshot name: %s",
                jobTypeParam.m_job.jobId.c_str(), item.m_snapshotId.c_str(), item.m_snapshotName.c_str());
            checkRet = false;
            break;
        }
    }
    return SUCCESS;
}

void KubernetesProtectEngine::DiscoverApplications(std::vector<Application>& /* returnValue */,
    const std::string& /* appType */)
{
}

void KubernetesProtectEngine::ListApplicationResource(
    std::vector<ApplicationResource>& /* returnValue */, const ApplicationEnvironment& /* appEnv */,
    const Application& /* application */, const ApplicationResource& /* parentResource */)
{
}

void KubernetesProtectEngine::DiscoverHostCluster(ApplicationEnvironment& /* returnEnv */,
    const ApplicationEnvironment& /* appEnv */)
{
}

void KubernetesProtectEngine::DiscoverAppCluster(ApplicationEnvironment &returnEnv,
                                                 const ApplicationEnvironment &appEnv, const Application &application)
{
    returnEnv.__set_id(appEnv.id);
    returnEnv.__set_name(appEnv.name);
    returnEnv.__set_type(appEnv.type);
    returnEnv.__set_subType(appEnv.subType);

    std::optional<std::vector<StorageParam>> storagesOpt = KubeHelper::GetStorageParamVecFromAppEnv(
        application.auth.extendInfo);
    if (!storagesOpt.has_value()) {
        ERRLOG("%s", EXTEND_INFO_INVALID);
        ThrowListFailException("Fail to extract authExtendInfo.");
    }

    Json::Value body;
    int errCode;
    for (const auto &item : storagesOpt.value()) {
        KubernetesPlugin::AccessAuthParam accessAuthParam(item.username, item.password, "0");
        auto[ret, storageClient]= KubeHelper::CheckStorageClientFromStorageParam(accessAuthParam, item);
        if (ret != SUCCESS) {
            std::string errorStr;
            StorageClient::CheckIpValidty(item.ip, item.port, item.ipList, accessAuthParam, errorStr);
            errCode = (ret == USERNAME_OR_PASSWORD_WRONG_V6 || ret == USERNAME_OR_PASSWORD_WRONG_V3) ?
                    STORAGE_AUTH_FAILED : STORAGE_NETWORK_FAILED;
            body["errorCode"] = errCode;
            body["errorIp"] = errorStr;
            body["message"] = DORADO_AUTH_FAIL;
            ERRLOG("%s Storage url: %s Storage username: %s ErrorCode: %d, errorStr: %s", DORADO_AUTH_FAIL.c_str(),
                item.ip.c_str(), item.username.c_str(), errCode, errorStr.c_str());
            break;
        }
        StorageDeviceInfo storageDeviceInfo;
        std::tie(ret, storageDeviceInfo) = storageClient->GetDeviceBaseInfo();
        if (ret != SUCCESS) {
            errCode = (ret == USERNAME_OR_PASSWORD_WRONG_V6 || ret == USERNAME_OR_PASSWORD_WRONG_V3) ?
                STORAGE_AUTH_FAILED : STORAGE_NETWORK_FAILED;
            body["errorCode"] = errCode;
            body["errorIp"] = item.ipList;
            body["message"] = DORADO_AUTH_FAIL;
            ERRLOG("Storage auth fail, ip: %s, port: %d, response code: %d", item.ip.c_str(), item.port, ret);
            break;
        }
        body[item.ip] = storageDeviceInfo.sn;
    }
    Json::FastWriter writer;
    returnEnv.__set_extendInfo(writer.write(body));
    INFOLOG("Start to discover cluster info, return sn info:%s", returnEnv.extendInfo.c_str());
}

void KubernetesProtectEngine::CheckApplication(ActionResult &_return, const ApplicationEnvironment &appEnv,
    const Application &application)
{
    // 1. 校验k8sApi输入参数，如果成功得到k8sApi, 并查看FlexVolume连通性
    auto kubernetesApi = KubeHelper::GetKubernetesApiFromAppEnv(appEnv.auth.extendInfo);
    if (!kubernetesApi) {
        ERRLOG("%s", EXTEND_INFO_INVALID);
        FillActionResult(_return, FAILED, CONNECT_FAILED, EXTEND_INFO_INVALID);
        return;
    }
    auto [ret, nameSpaceResourceList] = kubernetesApi->ListNameSpaces();
    // 2. 校验存储参数
    auto storageParamList = KubeHelper::GetStorageParamVecFromAppEnv(appEnv.auth.extendInfo);
    if (!storageParamList || storageParamList->size() == 0) {
        ERRLOG("%s", STORAGES_OR_CONFIG_EMPTY);
        FillActionResult(_return, FAILED, ERR_PARAM, STORAGES_OR_CONFIG_EMPTY);
        return;
    }
    std::set<std::string> storageUrlSet;
    std::tie(ret, storageUrlSet) = kubernetesApi->ListStorages();
    if (ret != 0) {
        ERRLOG("%s", K8S_API_CONNECT_FAIL);
        FillActionResult(_return, FAILED, CONNECT_FAILED, K8S_API_CONNECT_FAIL);
        return;
    }
    // 判断从K8S生产环境Pod对应的存储设备和注册下发的存储设备是否匹配
    if (storageUrlSet.size() != storageParamList->size()) {
        FillActionResult(_return, FAILED, STORAGE_NOT_MATCH, DORADO_AUTH_FAIL);
        return;
    }
    // 3. 连接存储
    if (CheckEveryStorage(_return, storageParamList.value(), storageUrlSet) == SUCCESS) {
        FillActionResult(_return, SUCCESS, SUCCESS, "OK");
    }
    return;
}

int32_t KubernetesProtectEngine::CheckEveryStorage(ActionResult &_return,
    const std::vector<StorageParam> &storageParamList, const std::set<std::string> &storageUrlSet)
{
    for (const auto &item: storageParamList) {
        std::vector<std::string> storageInputUrl = {};
        INFOLOG("Storage ip list: %s", item.ipList.c_str());
        item.GetStorageUrl(storageInputUrl);
        if (std::all_of(storageInputUrl.begin(), storageInputUrl.end(), [storageUrlSet]
            (std::string storageUrl) {return storageUrlSet.find(storageUrl) == storageUrlSet.end();})) {
            ERRLOG("Not verified url: %s", item.ip.c_str());
            FillActionResult(_return, FAILED, STORAGE_NOT_MATCH, DORADO_AUTH_FAIL);
            return FAILED;
        }
    }
    return SUCCESS;
}

void KubernetesProtectEngine::ListApplicationResourceV2(ResourceResultByPage &_return,
                                                        const ListResourceRequest &request)
{
    std::optional <KubernetesApi> kubernetesApiOpt = KubeHelper::GetKubernetesApiFromAppEnv(
        request.appEnv.auth.extendInfo);
    if (!kubernetesApiOpt.has_value()) {
        ERRLOG("AppEnv auth extendInfo parse error.");
        ThrowListFailException("Fail to extract authExtendInfo.");
    }

    _return.__set_pageNo(1);
    _return.__set_pageSize(0);
    _return.__set_pages(1);
    _return.__set_total(0);
    if (request.applications.empty()) {
        // 查询namespaces
        auto [ret, nameSpaceResourceList] = kubernetesApiOpt->ListNameSpaces();
        if (ret != 0) {
            ThrowListFailException("Fail to get namespaces from k8s cluster.");
        }
        for (auto &nameSpaceResource: nameSpaceResourceList) {
            nameSpaceResource.__set_parentId(request.appEnv.id);
        }
        _return.__set_items(nameSpaceResourceList);
        return;
    }

    Application application = request.applications.back();
    std::string subType = application.subType;
    if ((subType.compare("KubernetesNamespace")) == 0) {
        // 若是查询单个sts，则将需要查询的sts信息放到 application中的extendInfo中
        auto addStsRet = AddStsNameInApplicationExtendInfo(request, application);
        if (addStsRet != SUCCESS) {
            ERRLOG("Failed to input sts name into application extend info");
        }
        // 查询statefulSets
        auto [ret, stateFulSetResourceList] = kubernetesApiOpt->ListStatefulSet(application);
        if (ret != 0) {
            ThrowListFailException("Fail to get stateful sets from k8s cluster.");
        }
        _return.__set_items(stateFulSetResourceList);
    } else {
        DBGLOG("No more children now for subType: %s", subType.c_str());
    }
}

void KubernetesProtectEngine::FillActionResult(AppProtect::ActionResult &_return, int code, int errorCode,
    const std::string &message, const std::vector<std::string> &bodyErrParams)
{
    _return.__set_code(code);
    _return.__set_bodyErr(errorCode);
    _return.__set_message(message);
    _return.__set_bodyErrParams(bodyErrParams);
}

void KubernetesProtectEngine::ThrowListFailException(const std::string &message)
{
    AppProtect::AppProtectPluginException exception;
    exception.__set_code(FAILED);
    exception.__set_message(message);
    throw exception;
}

std::shared_ptr<AppProtect::BackupJob> KubernetesProtectEngine::GetBackupParam()
{
    std::shared_ptr<JobCommonInfo> jobCommonInfo = GetJobHandle()->GetJobCommonInfo();
    return std::dynamic_pointer_cast<AppProtect::BackupJob>(jobCommonInfo->GetJobInfo());
}

std::pair<int32_t, std::vector<std::string>> KubernetesProtectEngine::ParseVolumeNamesFromBackupParam(
    std::shared_ptr<AppProtect::BackupJob> &backupJob)
{
    std::vector<std::string> volumeNames = {};
    BackupJobParamExtendInfo backupJobParamExtendInfo;
    if (!Module::JsonHelper::JsonStringToStruct(backupJob->protectObject.extendInfo, backupJobParamExtendInfo)) {
        ERRLOG("Protect object extend info parse fail, protectObject.extendInfo: %s",
            backupJob->protectObject.extendInfo.c_str());
        return make_pair(FAILED, volumeNames);
    }

    Json::Value volumeNameJsonArray;
    Module::JsonHelper::JsonStringToJsonValue(backupJobParamExtendInfo.m_volumeNames, volumeNameJsonArray);
    for (const auto &item: volumeNameJsonArray) {
        INFOLOG("Ready to backup volume, temp name: %s", item.asString().c_str());
        if (std::find(volumeNames.begin(), volumeNames.end(), item.asString()) == volumeNames.end()) {
            volumeNames.push_back(item.asString());
        }
    }

    if (volumeNames.empty()) {
        INFOLOG("The backup volume names is empty.");
        return make_pair(FAILED, volumeNames);
    }

    return make_pair(SUCCESS, volumeNames);
}

int32_t KubernetesProtectEngine::FilterPodsByPreScript(const std::vector<ApplicationResource> &pods,
                                                       const std::shared_ptr<AppProtect::BackupJob> &backupParam,
                                                       std::vector<ApplicationResource> &backupPods)
{
    BackupJobParamExtendInfo backupJobParamExtendInfo;
    if (!Module::JsonHelper::JsonStringToStruct(backupParam->protectObject.extendInfo, backupJobParamExtendInfo)) {
        ERRLOG("Protect object extend info parse fail, protectObject.extendInfo: %s",
            backupParam->protectObject.extendInfo.c_str());
        return FAILED;
    }
    if (backupJobParamExtendInfo.m_preScript.empty()) {
        INFOLOG("FilterPodsByPreScript, but the pre-script path is empty.");
        backupPods = pods;
        return SUCCESS;
    }

    auto kubernetesApi = KubeHelper::GetKubernetesApiFromAppEnv(backupParam->protectEnv.auth.extendInfo);
    if (!kubernetesApi.has_value()) {
        ERRLOG("Failed to initialize kubernetesApi, protectObject.extendInfo: %s",
            backupParam->protectObject.extendInfo.c_str());
        return FAILED;
    }

    for (const auto &pod : pods) {
        PodExtendInfo podExtendInfo;
        if (!Module::JsonHelper::JsonStringToStruct(pod.extendInfo, podExtendInfo)) {
            ERRLOG("Struct PodExtendInfo to string fail.");
            continue;
        }

        bool isBackupEnable = false;
        for (const auto &containerName: podExtendInfo.m_containerNames) {
            std::string excResult = kubernetesApi->KubeExec(pod.parentName, pod.name, containerName,
                                                            backupJobParamExtendInfo.m_preScript);
            GenerateScriptLabel(excResult, pod.name, containerName, backupJobParamExtendInfo.m_preScript,
                                TaskStage::Pre_Task);
            isBackupEnable = KubeHelper::CheckKubeExecRetContainKey(excResult);
            if (isBackupEnable) {
                break;
            }
        }

        if (!isBackupEnable) {
            continue;
        }
        GenerateLabel(JobLogLevel::type::TASK_LOG_INFO, MARK_POD_BACKUP_SUCCESS_LABEL, std::vector{pod.name});
        backupPods.push_back(pod);
    }
    if (backupPods.empty()) {
        ERRLOG("FilterPodsByPreScript, all pods are not ready to backup.");
        GenerateLabel(JobLogLevel::type::TASK_LOG_ERROR, BACKUP_PODS_FILTER_EMPTY_LABEL, std::vector<std::string>{});
        return FAILED;
    }
    return SUCCESS;
}

int32_t KubernetesProtectEngine::FindOutTheBackupStorage(const std::vector<StorageParam> &storages,
                                                         std::vector<BatchSnapshotParam> &createParams)
{
    for (auto &createParam : createParams) {
        auto[isSameStorage, storageUrl] = KubeHelper::CheckPodPvsInSameStorage(createParam);
        if (!isSameStorage) {
            ERRLOG("Backup POD: %s, its pv list is not in the same storages.", createParam.m_pod.name.c_str());
            return FAILED;
        }
        bool isFound = false;
        for (const auto &storage: storages) {
            std::vector<std::string> storageUrlList = {};
            storage.GetStorageUrl(storageUrlList);
            if (count(storageUrlList.begin(), storageUrlList.end(), storageUrl)) {
                INFOLOG("Backup storage found successfully, storage url: %s", storageUrl.c_str());
                createParam.m_storage = storage;
                isFound = true;
                break;
            }
        }
        if (!isFound) {
            ERRLOG("Backup storage not found from backup param.");
            return FAILED;
        }
    }
    return SUCCESS;
}

int32_t KubernetesProtectEngine::TakeConsistentSnapshots(BatchSnapshotParam &batchSnapshotInfo,
                                                         std::map <std::string, ConsistentActivationInfo> &esnMap)
{
    auto [ret, storageClient] = KubeHelper::GetStorageClientFromStorageParam(batchSnapshotInfo.m_storage);
    if (ret != SUCCESS) {
        ERRLOG("Storage device authentication failed, ip:%s, username:%s, error code: %d",
               batchSnapshotInfo.m_storage.ip.c_str(), batchSnapshotInfo.m_storage.username.c_str(), ret);
        return FAILED;
    }
    StorageDeviceInfo storageDeviceInfo;
    std::tie(ret, storageDeviceInfo) = storageClient->GetDeviceBaseInfo();
    if (ret != SUCCESS) {
        ERRLOG("Failed to obtain basic information of village rough equipment, ip:%s, username:%s, error code: %d",
               batchSnapshotInfo.m_storage.ip.c_str(), batchSnapshotInfo.m_storage.username.c_str(), ret);
        return FAILED;
    }
    batchSnapshotInfo.m_storageDeviceInfo = storageDeviceInfo;
    std::vector <std::string> snapshotIds = {};
    std::vector <std::string> snapshotNames = {};
    std::vector <PerSnapshotParam> snapshotParams = {};
    time_t curtem = time(nullptr);
    unsigned long long timestamp = (unsigned long long) curtem;
    for (const auto &pv: batchSnapshotInfo.m_pvs) {
        int perRet = CreateSnapshot(pv, storageClient, timestamp, snapshotIds, snapshotParams);
        if (perRet != SUCCESS) {
            ERRLOG("Failed to create snapshot, LUN name: %s, error code: %d", pv.lunName.c_str(), perRet);
            return FAILED;
        }
    }
    batchSnapshotInfo.m_snapshotInfos = snapshotParams;
    for (auto ii = snapshotParams.begin(); ii != snapshotParams.end(); ++ii) {
        snapshotNames.push_back((*ii).m_snapshotInfoData.m_name);
    }
    auto iter = esnMap.find(storageDeviceInfo.sn);
    if (iter != esnMap.end()) {
        auto &consistentActivationInfo = iter->second;
        consistentActivationInfo.m_snapshotIds.insert(consistentActivationInfo.m_snapshotIds.end(), snapshotIds.begin(),
                                                      snapshotIds.end());
        consistentActivationInfo.m_snapshotNames.insert(consistentActivationInfo.m_snapshotNames.end(),
                                                        snapshotNames.begin(), snapshotNames.end());
    } else {
        ConsistentActivationInfo consistentActivationInfo;
        consistentActivationInfo.m_snapshotIds = snapshotIds;
        consistentActivationInfo.m_snapshotNames = snapshotNames;
        consistentActivationInfo.m_storage = batchSnapshotInfo.m_storage;

        esnMap[storageDeviceInfo.sn] = consistentActivationInfo;
    }
    time_t activeTem = time(NULL);
    unsigned long long activeTime = (unsigned long long) activeTem;
    batchSnapshotInfo.m_activeTime = activeTime;
    return SUCCESS;
}

int32_t KubernetesProtectEngine::ActiveConsistentSnapshots(std::map <std::string, ConsistentActivationInfo> &esnMap)
{
    // 对同一存储设备上的快照进行一致性激活
    for (auto &esnPair: esnMap) {
        auto storagePara = esnPair.second.m_storage;
        auto snapshotIds = esnPair.second.m_snapshotIds;
        auto snapshotNames = esnPair.second.m_snapshotNames;
        auto [ret, storageClient] = KubeHelper::GetStorageClientFromStorageParam(storagePara);
        if (ret != SUCCESS) {
            ERRLOG("Storage device authentication failed, ip:%s, username:%s, error code: %d",
                   storagePara.ip.c_str(), storagePara.username.c_str(), ret);
            return FAILED;
        }
        time_t activeTem = time(NULL);
        unsigned long long activeTime = (unsigned long long) activeTem;
        // 一次性激活快照
        ret = storageClient->ActivateSnapshot(snapshotIds);
        if (ret != SUCCESS) {
            ERRLOG("Failed to activate batch snapshots, error code: %d", ret);
            return FAILED;
        }
        std::string snapshotName;
        for (auto i = snapshotNames.begin(); i != snapshotNames.end(); ++i) {
            if (snapshotName.empty()) {
                snapshotName = *i;
            } else {
                snapshotName += ", " + (*i);
            }
        }
        GenerateLabel(JobLogLevel::type::TASK_LOG_INFO, ACTIVATE_SNAPS_SUCCESS_LABEL,
                      std::vector{KubeHelper::ConvertTimeStamp2TimeStr(activeTem), snapshotName});
    }
    return SUCCESS;
}

int32_t KubernetesProtectEngine::CreateSnapshot(const Pv &pv, std::shared_ptr<StorageClient> storageClient,
                                                unsigned long long timestamp, std::vector<std::string> &snapshotIds,
                                                std::vector<PerSnapshotParam> &snapshotParams)
{
    PerSnapshotParam perSnapshotParam;
    perSnapshotParam.m_pv = pv;
    // 通过卷名称获取卷ID
    auto[perRet, lunInfoData] = storageClient->GetLunInfoData(pv.lunName);
    if (perRet != SUCCESS) {
        ERRLOG("Failed to get volume information from storage device, LUN name: %s, error code: %d",
            pv.lunName.c_str(), perRet);
        return FAILED;
    }
    perSnapshotParam.m_lunInfoData = lunInfoData;
    // 根据卷ID打快照
    SnapshotCreateParam snapshotCreateParam;
    snapshotCreateParam.m_lunId = lunInfoData.m_id;
    snapshotCreateParam.m_name = GenerateSnapshotName(lunInfoData.m_id);
    snapshotCreateParam.m_description = m_snapDescription;
    SnapshotInfoData snapshotInfoData;
    std::tie(perRet, snapshotInfoData) = storageClient->CreateSnapshot(snapshotCreateParam);
    if (perRet != SUCCESS) {
        ERRLOG("Failed to create snapshot, LUN name: %s, LUN ID: %s, error code: %d",
            pv.lunName.c_str(), lunInfoData.m_id.c_str(), perRet);
        return FAILED;
    }
    perSnapshotParam.m_snapshotInfoData = snapshotInfoData;
    // 判断快照状态是否是已激活，若已激活，则取消激活
    if (snapshotInfoData.m_runningStatus == SNAPSHOT_ACTIVE_STATUS) {
        perRet = storageClient->StopSnapshot(snapshotInfoData.m_id);
        if (perRet != SUCCESS) {
            ERRLOG("Failed to stop snapshot, snapshot name: %s, snapshot ID: %s, error code: %d",
                snapshotInfoData.m_name.c_str(), snapshotInfoData.m_id.c_str(), perRet);
            return FAILED;
        }
    }
    snapshotIds.push_back(snapshotInfoData.m_id);
    snapshotParams.push_back(perSnapshotParam);
    return SUCCESS;
}

void KubernetesProtectEngine::GenerateVolSnapInfo(const BatchSnapshotParam &batchSnapshotInfo, SnapshotInfo &snapshot)
{
    for (const auto &item: batchSnapshotInfo.m_snapshotInfos) {
        SnapshotInfoData snapshotInfoData = item.m_snapshotInfoData;
        VirtPlugin::VolSnapInfo volSnapInfo;
        volSnapInfo.m_volUuid = item.m_lunInfoData.m_wwn;
        volSnapInfo.m_snapshotName = snapshotInfoData.m_name;
        volSnapInfo.m_snapshotId = snapshotInfoData.m_id;
        volSnapInfo.m_storageSnapId = snapshotInfoData.m_id;
        volSnapInfo.m_createTime = std::to_string(batchSnapshotInfo.m_activeTime);

        VirtPlugin::DatastoreInfo datastoreInfo;
        datastoreInfo.m_name = batchSnapshotInfo.m_storageDeviceInfo.name;
        datastoreInfo.m_moRef = batchSnapshotInfo.m_storageDeviceInfo.sn;
        datastoreInfo.m_type = batchSnapshotInfo.m_storageDeviceInfo.productMode;
        datastoreInfo.m_ip = batchSnapshotInfo.m_storage.ip;
        datastoreInfo.m_port = batchSnapshotInfo.m_storage.port;
        datastoreInfo.m_poolId = item.m_lunInfoData.m_parentId;
        datastoreInfo.m_ipList = batchSnapshotInfo.m_storage.ipList;

        VirtPlugin::VolumeDSExtendInfo dsExtend;
        dsExtend.m_volWwn = item.m_lunInfoData.m_wwn;
        dsExtend.m_volId = item.m_lunInfoData.m_id;
        dsExtend.m_volName = item.m_lunInfoData.m_name;
        Module::JsonHelper::StructToJsonString(dsExtend, datastoreInfo.m_extendInfo);

        volSnapInfo.m_datastore = datastoreInfo;
        std::string snapMetadata;
        Module::JsonHelper::StructToJsonString(snapshotInfoData, snapMetadata);
        volSnapInfo.m_metadata = snapMetadata;
        snapshot.m_volSnapList.push_back(volSnapInfo);
    }
}

int32_t KubernetesProtectEngine::FingerOutBackupPvs(const std::shared_ptr<AppProtect::BackupJob> &backupJob,
                                                    const std::vector<std::string> &volumeNames,
                                                    std::vector<BatchSnapshotParam> &createParams)
{
    auto kubernetesApi = KubeHelper::GetKubernetesApiFromAppEnv(backupJob->protectEnv.auth.extendInfo);
    if (!kubernetesApi.has_value()) {
        ERRLOG("Failed to initialize kubernetesApi, protectObject.extendInfo: %s",
            backupJob->protectObject.extendInfo.c_str());
        return FAILED;
    }

    Application statefulSet = backupJob->protectObject;
    auto [ret, pods] = kubernetesApi->ListPods(statefulSet);
    if (ret != SUCCESS) {
        ERRLOG("Fail to extract authExtendInfo.");
        return FAILED;
    }

    if (pods.empty()) {
        ERRLOG("Do not find any pod in charge under the stateful set, stateful set name: %s", statefulSet.name.c_str());
        return FAILED;
    }

    std::vector<ApplicationResource> backupPods;
    ret = FilterPodsByPreScript(pods, backupJob, backupPods);
    if (ret != SUCCESS) {
        ERRLOG("Find none pod to backup.");
        return FAILED;
    }
    for (const auto &backupPod : backupPods) {
        auto[perRet, pvs] = kubernetesApi->ListPvs(backupPod, volumeNames);
        if (perRet != SUCCESS) {
            ERRLOG("Parse backup pv list failed.");
            return FAILED;
        }
        if (pvs.size() < volumeNames.size()) {
            std::string volumeName = KubeHelper::FindLostVolumeNamesInPvs(volumeNames, pvs);
            ERRLOG("The number of PVs is less than the number of volumes, volume name:%s", volumeName.c_str());
            GenerateLabel(JobLogLevel::type::TASK_LOG_ERROR, FAILED_GET_PV_CORRESPONDING_TO_VOLUME_LABEL,
                std::vector<string>{volumeName});
            return FAILED;
        }
        BatchSnapshotParam batchSnapshotParam;
        batchSnapshotParam.m_pod = backupPod;
        batchSnapshotParam.m_pvs = pvs;
        createParams.push_back(batchSnapshotParam);
    }
    return SUCCESS;
}

int32_t KubernetesProtectEngine::ParseBackupPvsAndStorage(std::vector<BatchSnapshotParam> &createParams,
    const bool &labelFlag)
{
    std::shared_ptr<AppProtect::BackupJob> backupJob = GetBackupParam();

    auto[ret, volumeNames] = ParseVolumeNamesFromBackupParam(backupJob);
    if (ret != SUCCESS) {
        ERRLOG("Find none volume names to backup.");
        return FAILED;
    }

    ret = FingerOutBackupPvs(backupJob, volumeNames, createParams);
    if (ret != SUCCESS) {
        ERRLOG("Failed to finger out backup pv list.");
        return FAILED;
    }

    if (labelFlag) {
        for (const auto &createParam: createParams) {
            std::string lunNames = KubeHelper::JoinLunNames(createParam);
            GenerateLabel(JobLogLevel::type::TASK_LOG_INFO, SELECTED_BACKUP_VOLUME_LABEL,
                std::vector{createParam.m_pod.name, lunNames});
        }
    }

    auto storages = KubeHelper::GetStorageParamVecFromAppEnv(backupJob->protectEnv.auth.extendInfo);
    if (!storages.has_value()) {
        ERRLOG("Failed to parse and store backup information from backup parameters.");
        return FAILED;
    }

    ret = FindOutTheBackupStorage(storages.value(), createParams);
    if (ret != SUCCESS) {
        ERRLOG("Failed to find information about the storage device where the PV is located.");
        return FAILED;
    }
    return SUCCESS;
}
int KubernetesProtectEngine::GenVolPair(VMInfo &vmObj, const VolInfo &copyVol,
    const ApplicationResource &targetVol, VolMatchPairInfo &volPairs)
{
    if (m_restoreVolMap.find(copyVol.m_uuid) == m_restoreVolMap.end()) {
        ERRLOG("Can't find uuid's targetVol: %s", copyVol.m_uuid.c_str());
        return FAILED;
    }
    volPairs.AddVolPair(copyVol, m_restoreVolMap[copyVol.m_uuid]);
    return SUCCESS;
}

int32_t KubernetesProtectEngine::CheckAndReportStorage(const std::set<std::string> &overthresholdPoolNameSet,
    const int32_t &storageLimit, const BatchSnapshotParam &batchSnapshotInfo)
{
    if (!overthresholdPoolNameSet.empty()) {
        std::string overthresholdPoolNames;
        for (const std::string &pName : overthresholdPoolNameSet) {
            overthresholdPoolNames += pName;
            overthresholdPoolNames += ",";
        }
        overthresholdPoolNames.pop_back();
        VirtPlugin::ApplicationLabelType resourceCheckLabel;
        resourceCheckLabel.level = JobLogLevel::TASK_LOG_ERROR;
        resourceCheckLabel.label = SPACE_NOT_ABLE_LABEL;
        resourceCheckLabel.params = std::vector<std::string>{batchSnapshotInfo.m_storage.ip, overthresholdPoolNames,
            std::to_string(storageLimit)};
        ReportJobDetail(resourceCheckLabel);
        ERRLOG("Failed to CheckAndReportStorage, ip: %s, poolNames: %s, storageLimit: %d.",
            batchSnapshotInfo.m_storage.ip.c_str(), overthresholdPoolNames.c_str(), storageLimit);
        return FAILED;
    }
    return SUCCESS;
}

int32_t KubernetesProtectEngine::GetPoolIdByLunName(std::shared_ptr<StorageClient> storageClient,
    const std::string &lunName, StoragePoolData &poolData, std::map <std::string, std::set<std::string>> &poolMap)
{
    if (storageClient == nullptr) {
        ERRLOG("StorageClient nullptr");
        return FAILED;
    }
    auto [perRet, lunInfoData] = storageClient->GetLunInfoData(lunName);
    if (perRet != SUCCESS) {
        ERRLOG("Failed to get volume information from storage device, LUN name: %s, error code: %d",
            lunName.c_str(), perRet);
        return FAILED;
    }
    std::string poolId = lunInfoData.m_parentId;
    std::string sn = storageClient->GetStorageSnId();
    if (KubeHelper::FindStorageIdInMap(poolMap, poolId, sn)) {
        DBGLOG("The storage device(id: %s, sn: %s) where the PV is located has checked, pass.",
            poolId.c_str(), sn.c_str());
        return PASS;
    } else {
        auto poolRet = storageClient->QueryStoragePool(poolId, poolData);
        if (poolRet != SUCCESS) {
            ERRLOG("Failed to get pool information from storage device, pool id: %s, error code: %d",
                poolId.c_str(), poolRet);
            return FAILED;
        }
    }
    auto itPool = poolMap.find(sn);
    if (itPool == poolMap.end()) {
        poolMap[sn] = {poolId};
    } else {
        itPool->second.insert(poolId);
    }
    return SUCCESS;
}

int32_t KubernetesProtectEngine::CheckStorageThreshold(const BatchSnapshotParam &batchSnapshotInfo,
    std::map <std::string, std::set<std::string>> &poolMap, const int32_t &storageLimit)
{
    auto [ret, storageClient] = KubeHelper::GetStorageClientFromStorageParam(batchSnapshotInfo.m_storage);
    if (ret != SUCCESS || storageClient == nullptr) {
        ERRLOG("Storage device authentication failed, ip:%s, username:%s, error code: %d",
            batchSnapshotInfo.m_storage.ip.c_str(), batchSnapshotInfo.m_storage.username.c_str(), ret);
        return FAILED;
    }
    std::set<std::string> overthresholdPoolNameSet;
    for (const auto &pv: batchSnapshotInfo.m_pvs) {
        StoragePoolData poolData;
        int32_t iRet = GetPoolIdByLunName(storageClient, pv.lunName, poolData, poolMap);
        if (iRet == FAILED) {
            ERRLOG("storage pool(%s) usage is over limit(%d).", poolData.m_name.c_str(), storageLimit);
            return FAILED;
        }
        if (iRet == PASS || poolData.m_userTotalCapaity.empty() || poolData.m_userFreeCapacity.empty()) {
            INFOLOG("Pool pass. Lun:%s.", pv.lunName.c_str());
            continue;
        }
        DBGLOG("Pool capacity free:%s, total:%s, limit:%d.", poolData.m_userFreeCapacity.c_str(),
            poolData.m_userTotalCapaity.c_str(), storageLimit);
        if (boost::lexical_cast<uint64_t>(poolData.m_userTotalCapaity) != 0 &&
            (double(boost::lexical_cast<uint64_t>(poolData.m_userFreeCapacity)) /
            double(boost::lexical_cast<uint64_t>(poolData.m_userTotalCapaity))) * double(NUM_100) <
            (double(storageLimit))) {
            ERRLOG("storage pool(%s) usage is over limit(%d).", poolData.m_name.c_str(), storageLimit);
            if (overthresholdPoolNameSet.find(poolData.m_name) == overthresholdPoolNameSet.end()) {
                overthresholdPoolNameSet.insert(poolData.m_name);
            }
        }
    }
    return CheckAndReportStorage(overthresholdPoolNameSet, storageLimit, batchSnapshotInfo);
}

int32_t KubernetesProtectEngine::GetStorageLimit()
{
    std::shared_ptr<AppProtect::BackupJob> backupJob = GetBackupParam();
    if (backupJob != nullptr && !backupJob->extendInfo.empty()) {
        Json::Value extendInfo;
        if (!Module::JsonHelper::JsonStringToJsonValue(backupJob->extendInfo, extendInfo)) {
            ERRLOG("Trans job extend info to json value failed");
            return DEFAULT_STORAGE_LIMIT;
        }
        if (extendInfo.isMember("available_capacity_threshold")) {
            DBGLOG("Get available_capacity_threshold.");
            int32_t capacity = extendInfo["available_capacity_threshold"].asString().empty() ?
                DEFAULT_STORAGE_LIMIT : std::stoi(extendInfo["available_capacity_threshold"].asString());
            INFOLOG("Get available_capacity_threshold %d.", capacity);
            return capacity;
        }
    }
    return DEFAULT_STORAGE_LIMIT;
}

int32_t KubernetesProtectEngine::CheckBeforeBackup()
{
    INFOLOG("Start to CheckBeforeBackup.");
    std::vector <BatchSnapshotParam> createParams = {};
    int ret = ParseBackupPvsAndStorage(createParams);
    if (ret != SUCCESS) {
        ERRLOG("Parse pv list and storage fail.");
        return FAILED;
    }
    int32_t storageLimit = GetStorageLimit();
    std::map <std::string, std::set<std::string>> poolMap;  // ESN : storageId list
    for (auto &createParam: createParams) {
        ret = CheckStorageThreshold(createParam, poolMap, storageLimit) == SUCCESS ? ret : FAILED;
        if (ret != SUCCESS) {
            ERRLOG("CheckStorageThreshold(%s) failed.", createParam.m_storage.ip.c_str());
        }
    }
    return ret;
}

int32_t KubernetesProtectEngine::CheckBeforeRecover(const VMInfo &vmObj)
{
    // 1. 得到RestoreJob，做简单的校验
    std::shared_ptr<AppProtect::RestoreJob> restoreJobPtr = GetRestoreParam();

    // 2. 在所有卷中运行目标Pod中的前置脚本
    auto preScriptOpt = KubeHelper::GetScriptFromExtendInfo(restoreJobPtr->targetObject.extendInfo,
        ScriptType::PRE_SCRIPT);
    if (preScriptOpt && !preScriptOpt.value().empty()) {
        INFOLOG("Ready to Run preTask script: %s", preScriptOpt.value().c_str());
        RunScriptOnAllPodsInTheStateFulSet(preScriptOpt.value(), restoreJobPtr, TaskStage::Pre_Task);
    }

    // 3. 获取目标卷的卷列表VolNameList
    auto storageParamList = KubeHelper::GetStorageParamVecFromAppEnv(restoreJobPtr->targetEnv.auth.extendInfo);
    if (!storageParamList || storageParamList->size() == 0) {
        return FAILED;
    }

    if (restoreJobPtr->restoreSubObjects.empty()) {
        INFOLOG("There are no volumes to be restored! Wrong params!");
        return FAILED;
    }
    for (const auto &volResource : restoreJobPtr->restoreSubObjects) {
        auto storageParamOpt = KubeHelper::GetStorageParamFromVol(*storageParamList, volResource.extendInfo);
        if (!storageParamOpt) {
            return FAILED;
        }
        StorageParam storageParam = storageParamOpt.value();
        auto [ret, storageClient] = KubeHelper::GetStorageClientFromStorageParam(storageParam);
        if (ret != SUCCESS) {
            return FAILED;
        }
        auto targetVolOpt = KubeHelper::GenTargetVolFromInfos(volResource, storageParam, storageClient);
        if (!targetVolOpt) {
            ERRLOG("Check Failed! No matched volumeName: %s", volResource.name.c_str());
            return FAILED;
        }
        m_restoreVolMap[volResource.id] = targetVolOpt.value();
    }
    return SUCCESS;
}

std::shared_ptr<AppProtect::RestoreJob> KubernetesProtectEngine::GetRestoreParam()
{
    std::shared_ptr<JobCommonInfo> jobCommonInfo = GetJobHandle()->GetJobCommonInfo();
    return std::dynamic_pointer_cast<AppProtect::RestoreJob>(jobCommonInfo->GetJobInfo());
}

void KubernetesProtectEngine::GenerateLabel(JobLogLevel::type level, const std::string &label,
    const std::vector<std::string> &params)
{
    VirtPlugin::ApplicationLabelType applicationLabelType;
    applicationLabelType.level = level;
    applicationLabelType.label = label;
    applicationLabelType.params = params;
    ReportJobDetail(applicationLabelType);
}

bool KubernetesProtectEngine::GenerateScriptLabel(const std::string &excResult, const std::string &podName,
                                                  const std::string &containerName, const std::string &script,
                                                  TaskStage stage)
{
    auto [isSuccess, label] = KubeHelper::GetLabelByExecKubeResult(excResult, stage);
    JobLogLevel::type level;
    std::string preScriptExecRes;
    switch (stage) {
        case TaskStage::Pre_Task:
            level = isSuccess ? JobLogLevel::type::TASK_LOG_INFO : JobLogLevel::type::TASK_LOG_ERROR;
            // 执行脚本成功后，需要打印脚本执行的回显信息，且 回显结果包含 failed，则标签状态置为error
            if (isSuccess) {
                preScriptExecRes = KubeHelper::GetKubePreScriptExecResult(excResult);
                if (preScriptExecRes.find("failed") != std::string::npos) {
                    level = JobLogLevel::type::TASK_LOG_ERROR;
                }
            }
            break;
        case TaskStage::Post_Task:
            level = isSuccess ? JobLogLevel::type::TASK_LOG_INFO : JobLogLevel::type::TASK_LOG_WARNING;
            break;
        case TaskStage::Fail_Task:
            level = isSuccess ? JobLogLevel::type::TASK_LOG_INFO : JobLogLevel::type::TASK_LOG_WARNING;
            break;
        default:
            break;
    }

    DBGLOG("Generate log info, level: %d label: %s podName: %s containerName: %s script: %s",
           level, label.c_str(), podName.c_str(), containerName.c_str(), script.c_str());
    GenerateLabel(level, label, std::vector{podName, containerName, script, preScriptExecRes});
}

int32_t KubernetesProtectEngine::GetSnapshotsOfVolume(const VolInfo &volInfo,
                                                      std::vector<VirtPlugin::VolSnapInfo> &snapList)
{
    std::shared_ptr<AppProtect::BackupJob> backupParam = GetBackupParam();
    auto storages = KubeHelper::GetStorageParamVecFromAppEnv(backupParam->protectEnv.auth.extendInfo);
    if (!storages.has_value()) {
        ERRLOG("Failed to parse and store backup information from backup parameters.");
        return FAILED;
    }
    std::string storageSn = volInfo.m_datastore.m_moRef;
    if (storageSn.empty()) {
        ERRLOG("Failed to get sn from volInfo.");
        return FAILED;
    }
    auto storageOpt = KubeHelper::FindMatchSnStorage(storages.value(), storageSn);
    if (!storageOpt.has_value()) {
        ERRLOG("No usable storage device param found in backup param, required sn: %s, taskId: %s",
            storageSn.c_str(), backupParam->jobId.c_str());
        return FAILED;
    }
    StorageParam storage = storageOpt.value();
    auto[initRet, storageClient] = KubeHelper::GetStorageClientFromStorageParam(storage);
    if (initRet != SUCCESS) {
        ERRLOG("Storage auth fail, taskId: : %s, ip: %s, port: %s, response code: %s",
            backupParam->jobId.c_str(), storage.ip.c_str(), std::to_string(storage.port).c_str(),
            std::to_string(initRet).c_str());
        return FAILED;
    }
    auto[perRet, snapshotResponse] = storageClient->GetLunSnapshots(volInfo.m_name);
    if (perRet == FAILED) {
        ERRLOG("Get lun snapshots failed, volId: %s", volInfo.m_uuid.c_str());
        return FAILED;
    }
    for (auto &snap : snapshotResponse.m_SnapshotInfoDatas) {
        DBGLOG("Get lun snapshots ESN: %s.", m_xNNEsn.c_str());
        if (!MatchSnapshotName(snap.m_name) || !MatchSnapshotDescription(snap.m_description)) {
            continue;
        }
        VirtPlugin::VolSnapInfo volSnap;
        volSnap.m_volUuid = volInfo.m_uuid;
        volSnap.m_snapshotName = snap.m_name;
        volSnap.m_snapshotId = snap.m_id;
        volSnap.m_storageSnapId = snap.m_id;
        volSnap.m_datastore = volInfo.m_datastore;
        std::string snapMetadata;
        Module::JsonHelper::StructToJsonString(snap, snapMetadata);
        volSnap.m_metadata = snapMetadata;
        snapList.push_back(volSnap);
    }
    return SUCCESS;
}

int32_t KubernetesProtectEngine::AddStsNameInApplicationExtendInfo(const ListResourceRequest &request,
                                                                   Application &application)
{
    if (!request.condition.conditions.empty()) {
        Json::Value extendInfo;
        if (!Module::JsonHelper::JsonStringToJsonValue(request.condition.conditions, extendInfo)) {
            ERRLOG("Trans json value Application extendInfo failed, %s", request.condition.conditions.c_str());
            return FAILED;
        }
        if (extendInfo.isMember("statefulSet")) {
            application.extendInfo = extendInfo["statefulSet"].asString();
            INFOLOG("statefulSet name provided for get single sts: %s", extendInfo["statefulSet"].asString().c_str());
        } else {
            ERRLOG("No advanceParams provided for getting sts, statefulSet is required");
            return FAILED;
        }
    }
    return SUCCESS;
}

int32_t KubernetesProtectEngine::CheckBeforeMount()
{
    return SUCCESS;
}

int32_t KubernetesProtectEngine::CancelLiveMount(const VMInfo &liveVm)
{
    return SUCCESS;
}

int32_t KubernetesProtectEngine::CreateLiveMount(const VMInfo &copyVm, VMInfo &newVm)
{
    return SUCCESS;
}
}